package com.springboot.component;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;

@Component("rocketMqProducer")
public class RocketMqProducer implements InitializingBean, DisposableBean {

    private static final Logger log = LoggerFactory.getLogger(RocketMqProducer.class);

    @Value("${rocketmq.producerGroupName}")
    private String productGroup = null;
    @Value("${rocketmq.producerNamesrvAddr}")
    private String addrList = null;
    @Value("${rocketmq.test.topic}")
    private String topic = null;

    public DefaultMQProducer rocketMqProducer = null;

    public volatile boolean initiated = false;

    public void init(){
        rocketMqProducer = new DefaultMQProducer(productGroup);
        rocketMqProducer.setNamesrvAddr(addrList);
        try {
            rocketMqProducer.start();
            initiated = true;
        } catch (MQClientException e) {}
    }

    public void sendMsg(String tag, String msgBody) {
        try {
            Message msg = new Message(topic, tag, msgBody.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = rocketMqProducer.send(msg);
            System.out.println("msgId:" + sendResult.getMsgId());
            System.out.printf("%s%n", sendResult);
            System.out.printf("%s%n", "end line");
        } catch (UnsupportedEncodingException e){
            e.printStackTrace();
        }catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void sendMsg(String tagName, byte[] msgBody){//,int level
        Message msg = new Message(topic,tagName,msgBody);
        try {
            while(!initiated){
                Thread.sleep(1000L);
            }
            SendResult result = rocketMqProducer.send(msg);
            log.info(result.toString());
        } catch (Exception e) {
            log.error("msgBody:"+new String(msgBody), e);
        }
    }

    public void sendMsg(String tag, byte[] msgBody, String topic) {
        Message msg = new Message(topic, tag, msgBody);
        try {
            SendResult result = rocketMqProducer.send(msg);
            log.info("消息发送回执：{}",result);
        } catch (Exception e) {
            log.error("msgBody:"+new String(msgBody), e);
        }


    }

    public String sendMsg(String topic,String tag, byte[] msgBody) {
        Message msg = new Message(topic, tag, msgBody);
        try {
            SendResult result = rocketMqProducer.send(msg);
            log.info("消息发送回执：{}",result);
            return result.getMsgId();
        } catch (Exception e) {
            log.error("msgBody:"+new String(msgBody), e);
        }
        return null;
    }

    @Override
    public void destroy() throws Exception {
        rocketMqProducer.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        init();
    }
}
